package job.mfp.document_builder;

import java.io.IOException;

import mapred.example.ParseReader;
import mapred.example.model.Anchor;
import mapred.example.model.Fields;
import mapred.example.model.Link;
import mapred.example.model.Page;

import org.apache.hadoop.io.Text;

public class MFP_DocumentBuilder_Mapper extends ParseReader {

    @Override
    public void map(Page page) throws IOException {
        String key = page.getUrl().getNormalized();

        this.output(new Text(key), new Text(Fields.URL_SEG + "\t"
                + page.encodeUrl()));
        this.output(new Text(key), new Text(Fields.TITLE + "\t"
                + page.encodeTitle()));
        this.output(new Text(key), new Text(Fields.BODY + "\t"
                + page.encodeBody()));
        for (Link link : page.getOutLinks()) {
            Anchor anchor = Anchor.newInstanceFromLink(link);
            this.output(new Text(link.getUrl()), new Text(Fields.ANCHOR + "\t"
                    + anchor.encode()));
        }

        this.incrCounter("Counter", "Document", 1L);
    }

    @Override
    public void setUp() {
    }

    @Override
    public void tearDown() throws IOException {
    }

}
